// Copyright (c) 2023 Institute of Software, Chinese Academy of Sciences
// Ripple is licensed under Mulan PSL v2.
// You can use this software according to the terms and conditions of the Mulan PSL v2.
// You may obtain a copy of Mulan PSL v2 at:
//          http://license.coscl.org.cn/MulanPSL2
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
// See the Mulan PSL v2 for more details.
package ripple.server.core;

import java.io.FileWriter;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

import javax.servlet.Servlet;

import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.Channel;
import oshi.SystemInfo;
import oshi.hardware.CentralProcessor;
import oshi.hardware.HardwareAbstractionLayer;
import ripple.common.ClientListCache;
import ripple.common.DebugTool;
import ripple.common.entity.AbstractMessage;
import ripple.common.entity.ClientMetadata;
import ripple.common.entity.Constants;
import ripple.common.entity.Data;
import ripple.common.entity.DataMessage;
import ripple.common.entity.DeleteMessage;
import ripple.common.entity.IncrementalUpdateMessage;
import ripple.common.entity.Item;
import ripple.common.entity.NodeMetadata;
import ripple.common.entity.Reform;
import ripple.common.entity.ReformMessage;
import ripple.common.entity.Towards;
import ripple.common.entity.UpdateMessage;
import ripple.common.storage.Storage;
import ripple.server.core.dispatcher.ClientDispatcher;
import ripple.server.core.dispatcher.EqualDivisionClientDispatcher;
import ripple.server.core.overlay.FicusOverlay;
import ripple.server.core.overlay.Overlay;
import ripple.server.core.overlay.ReformOverlay;
import ripple.server.helper.Api;
import ripple.server.tcp.NettyServer;
import ripple.server.ui.AddConfigServlet;
import ripple.server.ui.ClientClusterServlet;
import ripple.server.ui.Endpoint;
import ripple.server.ui.GetConfigServlet;
import ripple.server.ui.GetSubscriptionServlet;
import ripple.server.ui.HomeServlet;
import ripple.server.ui.IncrementalUpdateServlet;
import ripple.server.ui.ModifyConfigServlet;
import ripple.server.ui.RemoveConfigServlet;
import ripple.server.ui.ServerClusterServlet;
import ripple.server.ui.StyleServlet;

/**
 * @author Zhen Tang
 */
public class Node {

    private static final Logger LOGGER = LoggerFactory.getLogger(Node.class);

    private ExecutorService executorService;
    private Worker worker;
    private Thread workingThread;

    private int id;
    private Overlay overlay;
    private Storage storage;
    private List<NodeMetadata> nodeList;
    private ConcurrentHashMap<Item, Set<ClientMetadata>> subscription;

    public ConcurrentHashMap<String, Set<ClientMetadata>> getMQSubscription() {
        return MQSubscription;
    }

    public void setMQSubscription(ConcurrentHashMap<String, Set<ClientMetadata>> MQSubscription) {
        this.MQSubscription = MQSubscription;
    }

    private ConcurrentHashMap<String, Set<ClientMetadata>> MQSubscription;
    private Set<ClientMetadata> connectedClients;

    private Tracker tracker;
    private HealthManager healthManager;
    private ClientListCache clientListCache;
    private ClientDispatcher clientDispatcher;

    private String address;
    private int uiPort;
    private int apiPort;
    private Server uiServer;
    private NettyServer apiServer;
    private boolean running;

    private double currentCpuLoad;

    private int responseCount;

    private List<NodeMetadata> newNodeList;

    public List<NodeMetadata> getNewNodeList() {
        return this.newNodeList;
    }

    public void setNewNodeList(List<NodeMetadata> nodeList) {
        this.newNodeList = nodeList;
    }

    public String getState() {
        return state;
    }

    public void setState(String state) {
        this.state = state;
    }

    private String state;

    public int getOverlayTerm() {
        return overlayTerm;
    }

    public void setOverlayTerm(int overlayTerm) {
        this.overlayTerm = overlayTerm;
    }

    private int overlayTerm;

    public ExecutorService getExecutorService() {
        return executorService;
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public Worker getWorker() {
        return worker;
    }

    private void setWorker(Worker worker) {
        this.worker = worker;
    }

    public Thread getWorkingThread() {
        return workingThread;
    }

    private void setWorkingThread(Thread workingThread) {
        this.workingThread = workingThread;
    }

    public int getId() {
        return id;
    }

    private void setId(int id) {
        this.id = id;
    }

    public Overlay getOverlay() {
        return overlay;
    }

    public void setOverlay(Overlay overlay) {
        this.overlay = overlay;
    }

    public Storage getStorage() {
        return storage;
    }

    private void setStorage(Storage storage) {
        this.storage = storage;
    }

    public List<NodeMetadata> getNodeList() {
        return nodeList;
    }

    public void setNodeList(List<NodeMetadata> nodeList) {
        this.nodeList = nodeList;
    }

    public ConcurrentHashMap<Item, Set<ClientMetadata>> getSubscription() {
        return subscription;
    }

    public void setSubscription(ConcurrentHashMap<Item, Set<ClientMetadata>> subscription) {
        this.subscription = subscription;
    }

    public Set<ClientMetadata> getConnectedClients() {
        return connectedClients;
    }

    public void setConnectedClients(Set<ClientMetadata> connectedClients) {
        this.connectedClients = connectedClients;
    }

    public Tracker getTracker() {
        return tracker;
    }

    private void setTracker(Tracker tracker) {
        this.tracker = tracker;
    }

    public HealthManager getHealthManager() {
        return healthManager;
    }

    private void setHealthManager(HealthManager healthManager) {
        this.healthManager = healthManager;
    }

    public ClientListCache getClientListCache() {
        return clientListCache;
    }

    public void setClientListCache(ClientListCache clientListCache) {
        this.clientListCache = clientListCache;
    }

    public ClientDispatcher getClientDispatcher() {
        return clientDispatcher;
    }

    public void setClientDispatcher(ClientDispatcher clientDispatcher) {
        this.clientDispatcher = clientDispatcher;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public int getUiPort() {
        return uiPort;
    }

    public void setUiPort(int uiPort) {
        this.uiPort = uiPort;
    }

    public int getApiPort() {
        return apiPort;
    }

    public void setApiPort(int apiPort) {
        this.apiPort = apiPort;
    }

    public Server getUiServer() {
        return uiServer;
    }

    public void setUiServer(Server uiServer) {
        this.uiServer = uiServer;
    }

    public NettyServer getApiServer() {
        return apiServer;
    }

    public void setApiServer(NettyServer apiServer) {
        this.apiServer = apiServer;
    }

    public boolean isRunning() {
        return running;
    }

    public void setRunning(boolean running) {
        this.running = running;
    }

    public double getCurrentCpuLoad() {
        return currentCpuLoad;
    }

    public void setCurrentCpuLoad(double currentCpuLoad) {
        this.currentCpuLoad = currentCpuLoad;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    private String type;

    public int getGroupId() {
        return groupId;
    }

    public void setGroupId(int groupId) {
        this.groupId = groupId;
    }

    private int groupId;

//    public Storage getMQStorage() {
//        return MQStorage;
//    }
//
//    public void setMQStorage(Storage MQStorage) {
//        this.MQStorage = MQStorage;
//    }
//
//    private Storage MQStorage;
    public List<Reform> getReformList() {
        return reformList;
    }

    public void setReformList(List<Reform> reformList) {
        this.reformList = reformList;
    }

    private List<Reform> reformList;

    public int getReformLeaderId() {
        return reformLeaderId;
    }

    public void setReformLeaderId(int reformLeaderId) {
        this.reformLeaderId = reformLeaderId;
    }

    private int reformLeaderId;

    public Date getLastReformUpdate() {
        return lastReformUpdate;
    }

    public void setLastReformUpdate(Date lastReformUpdate) {
        this.lastReformUpdate = lastReformUpdate;
    }

    private Date lastReformUpdate;

    public Overlay getNewOverlay() {
        return newOverlay;
    }

    public void setNewOverlay(Overlay newOverlay) {
        this.newOverlay = newOverlay;
    }

    private Overlay newOverlay;

    private Date proposeDate;

    public void setProposeDate(Date date) {
        this.proposeDate = date;
    }

    public Date getProposeDate() {
        return this.proposeDate;
    }

    public Node(int id, Overlay overlay, String storageLocation) {
        this(id, overlay, storageLocation, 0, 0);
    }

    public Node(int id, Overlay overlay, String storageLocation, int apiPort, int uiPort) {
        this.setExecutorService(Executors.newCachedThreadPool());
        this.setId(id);
        this.setOverlay(overlay);
        this.setTracker(new Tracker(this, overlay));
        this.setStorage(new Storage(storageLocation));
        this.setHealthManager(new HealthManager(this));
        this.setWorker(new Worker(this));

        this.setNodeList(new ArrayList<>());
        this.setSubscription(new ConcurrentHashMap<>());
        this.setApiPort(apiPort);
        this.setUiPort(uiPort);
        this.setConnectedClients(new HashSet<>());
        this.setClientListCache(new ClientListCache());
        this.setClientDispatcher(new EqualDivisionClientDispatcher(this));
        this.updateCpuLoad(1000);

        this.setMessageQueue(new ConcurrentHashMap<>());
        this.setMQSubscription(new ConcurrentHashMap<>());
//        this.setMQStorage(new Storage(storageLocation));

        // this.setState(NORMAL_STATE);
        this.setOverlayTerm(0);
        this.setReformList(null);
        this.setReformLeaderId(-1);
        this.setStorage(new Storage(storageLocation));
    }

    private void registerServlet(ServletContextHandler servletContextHandler, Servlet servlet, String endpoint) {
        servletContextHandler.addServlet(new ServletHolder(servlet), endpoint);
    }

    public void registerHandlers(ServletContextHandler servletContextHandler) {
        // UI
        this.registerServlet(servletContextHandler, new HomeServlet(this), Endpoint.UI_HOME);
        this.registerServlet(servletContextHandler, new StyleServlet(this), Endpoint.UI_STYLE);
        this.registerServlet(servletContextHandler, new GetConfigServlet(this), Endpoint.UI_GET_CONFIG);
        this.registerServlet(servletContextHandler, new AddConfigServlet(this), Endpoint.UI_ADD_CONFIG);
        this.registerServlet(servletContextHandler, new ModifyConfigServlet(this), Endpoint.UI_MODIFY_CONFIG);
        this.registerServlet(servletContextHandler, new IncrementalUpdateServlet(this), Endpoint.UI_INCREMENTAL_UPDATE);
        this.registerServlet(servletContextHandler, new RemoveConfigServlet(this), Endpoint.UI_REMOVE_CONFIG);
        this.registerServlet(servletContextHandler, new GetSubscriptionServlet(this), Endpoint.UI_GET_SUBSCRIPTION);
        this.registerServlet(servletContextHandler, new ServerClusterServlet(this), Endpoint.UI_SERVER_CLUSTER);
        this.registerServlet(servletContextHandler, new ClientClusterServlet(this), Endpoint.UI_CLIENT_CLUSTER);
    }

    public Item get(String applicationName, String key) {
        return this.getStorage().getItemService().getItem(applicationName, key);
    }

    public List<Item> getAll() {
        return this.getStorage().getItemService().getAllItems();
    }

    public boolean put(String applicationName, String key, String value) {
        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();

        UpdateMessage message = new UpdateMessage(applicationName, key, value, lastUpdate, lastUpdateServerId);
        message.setFromWhere(this.getId());

        this.propagateMessage(message);

        return true;
    }

    public boolean push(String topic, Data data) {
        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();

        DataMessage message = new DataMessage(topic, data, lastUpdate, lastUpdateServerId);

        message.setTowards(Towards.E2C);

        this.propagateMQMessage(message);

        return true;
    }

    public boolean reform() {

        if (!(this.getOverlay() instanceof ReformOverlay)) {
            return false;
        }

        if (!ReformOverlay.CONTINUE.equals(((ReformOverlay) this.getNewOverlay()).reform(this.getId()))) {
            return false;
        }

        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();

        int newTerm = ((ReformOverlay) this.getNewOverlay()).getTerm() + 1;
        List<NodeMetadata> newNodeList = this.getAliveNodeList();

        ReformMessage message = new ReformMessage(
                ReformOverlay.PROPOSE,
                ReformOverlay.getNewOverlayString(newTerm, newNodeList),
                lastUpdate, lastUpdateServerId);
        this.propagateMessage(message);

        return true;

    }

    public List<NodeMetadata> getAliveNodeList() {
        List<NodeMetadata> aliveNodeList = new ArrayList<>();
        for (NodeMetadata nodeMetadata : this.getNodeList()) {
            if (this.getHealthManager().isAlive(nodeMetadata)) {
                aliveNodeList.add(nodeMetadata);
            }
        }
        return aliveNodeList;
    }

    // public boolean prepareReformOverLay(Reform reform) {
    //     Date lastUpdate = new Date(System.currentTimeMillis());
    //     int lastUpdateServerId = this.getId();
    //     if (Objects.equals(this.getState(), NORMAL_STATE)) {
    //         ReformRequest message = ReformRequest.prepare(lastUpdate, lastUpdateServerId);
    //         this.setState(PROPOSER_STATE);
    //         this.setReformList(new ArrayList<>());
    //         this.getReformList().add(reform);
    //         this.setReformLeaderId(this.getId());
    //         this.propagateReformMessage(message);
    //         this.getExecutorService().submit(new Callable<Void>() {
    //             @Override
    //             public Void call() throws Exception {
    //                 try {
    //                     Thread.sleep(Reform.MAX_WAIT_TIME);
    //                     proposeReformOverLay(message);
    //                 } catch (Exception exception) {
    //                     exception.printStackTrace();
    //                 }
    //                 return null;
    //             }
    //         });
    //     } else {
    //         this.retransmit(reform);
    //     }
    //     return true;
    // }
    private void proposeReformOverLay() {
        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();
//         if (Objects.equals(this.getState(), PROPOSER_STATE)) {
// //            ReformRequest message = ReformRequest.prepare(lastUpdate, lastUpdateServerId);
//             this.setState(LEADER_STATE);
// //            this.setReformList(new ArrayList<>());
//             List<NodeMetadata> newNodeList = Reform.merge(this.getReformList());
//             this.setReformList(null);

//             Class<Overlay> overlayClass = (Class<Overlay>) this.getOverlay().getClass();
//             this.newOverlay = overlayClass.newInstance();
//             this.propagateReformMessage(message);
//             this.getExecutorService().submit(new Callable<Void>() {
//                 @Override
//                 public Void call() throws Exception {
//                     try {
//                         Thread.sleep(Reform.MAX_WAIT_TIME);
// //                        commitReformOverLay();
//                     } catch (Exception exception) {
//                         exception.printStackTrace();
//                     }
//                     return null;
//                 }
//             });
//         }
//        return true;
    }

//    private void proposeReformOverLay(AbstractMessage prepareMessage) {
//        Set<Integer> nodeIdSet = this.getStorage().getAckService().getAck(prepareMessage.getUuid()).getNodeList();
//        List<NodeMetadata> nodeMetadataList = new ArrayList<>();
//        for (Integer id : nodeIdSet) {
//            nodeMetadataList.add(this.getNodeById(id));
//        }
//        for (Reform reform : this.getReformList()) {
//            if (reform.getType().equals(Reform.JOIN)) {
//                nodeMetadataList.add(reform.getNodeMetadataList().get(0));
//            } else if (reform.getType().equals(Reform.LEAVE)) {
//                nodeMetadataList.remove(reform.getNodeMetadataList().get(0));
//            }
//        }
//
//        try {
//            this.setNewOverlay(this.getOverlay().getClass().newInstance());
//            this.getNewOverlay().buildOverlay(nodeMetadataList);
//        } catch (InstantiationException | IllegalAccessException e) {
//            throw new RuntimeException(e);
//        }
//
//        Date lastReformUpdate = this.getLastReformUpdate();
//        ReformRequest proposeMessage = new ReformRequest(lastUpdate, lastUpdateServerId);
//        this.setState(LEADER_STATE);
    ////        this.setReformLeaderId(this.getId());
//        this.propagateReformMessage(proposeMessage);
//
//
//    }

    private NodeMetadata getNodeById(int id) {
        for (NodeMetadata node : this.getNodeList()) {
            if (node.getId() == id) {
                return node;
            }
        }
        return null;
    }

    private void retransmit(Reform reform) {
    }

    public boolean incrementalUpdate(String applicationName, String key, UUID baseMessageUuid, String atomicOperation, String value) {
        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();

        IncrementalUpdateMessage message = new IncrementalUpdateMessage(applicationName, key,
                baseMessageUuid, atomicOperation, value, lastUpdate, lastUpdateServerId);
        this.propagateMessage(message);

        return true;
    }

    public boolean delete(String applicationName, String key) {
        Date lastUpdate = new Date(System.currentTimeMillis());
        int lastUpdateServerId = this.getId();

        DeleteMessage message = new DeleteMessage(applicationName, key, lastUpdate, lastUpdateServerId);
        this.propagateMessage(message);

        return true;
    }

    public boolean propagateMessage(final AbstractMessage message) {
        // Update local storage
        this.applyMessageToStorage(message);

        message.setFromWhere(this.getId());

        // reform overlay
        if (message instanceof ReformMessage) {
            synchronized (this.getOverlay()) {
                String operation = ((ReformOverlay) this.getOverlay()).reformMessageHandler((ReformMessage) message, this.getId());
                switch (operation) {
                    case ReformOverlay.REJECT:
                        AbstractMessage rejectMessage = ((ReformOverlay) this.getOverlay()).getRejectMessage((ReformMessage) message, this.getId());
                        this.sendTo(rejectMessage, message.getLastUpdateServerId());
                        return false;
                    case ReformOverlay.CONTINUE:
                        break;
                    case ReformOverlay.APPLY_COMMIT:
                        ((ReformOverlay) this.getOverlay()).applyCommit(ReformOverlay.getNewOverlayNodeListByString(((ReformMessage) message).getKey()));

                        break;
                    case ReformOverlay.APPLY_DEPRECATE:
                        ((ReformOverlay) this.getOverlay()).applyDeprecate();
                        break;
                    default:
                        break;
                }
            }
        }

        // Init ACK if it is the message source
        if (message.getLastUpdateServerId() == this.getId()) {
            this.getTracker().initProgress(message);
        }

        // Notify clients
        this.doNotifyClients(message);

        final int sourceId = message.getLastUpdateServerId();
        final int currentId = this.getId();

        // this.doSyncWithServer(message, sourceId, currentId);
        this.getExecutorService().submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                try {
                    doSyncWithServer(message, sourceId, currentId);
                    return null;
                } catch (Exception exception) {
                    exception.printStackTrace();
                    return null;
                }
            }
        });

        return true;
    }

    public void sendTo(AbstractMessage message, int targetId) {
        NodeMetadata targetNodeMetadata = null;

        for (NodeMetadata nodeMetadata : this.getNodeList()) {
            if (nodeMetadata.getId() == targetId) {
                targetNodeMetadata = nodeMetadata;
                break;
            }
        }

        if (this.getHealthManager().isAlive(targetNodeMetadata)) {
            LOGGER.info("[Node-{}] Sync {} with server {}:{}.", this.getId(), message.getType(), targetNodeMetadata.getAddress(), targetNodeMetadata.getPort());
            Channel channel = this.getApiServer().findChannel(targetNodeMetadata.getAddress(), targetNodeMetadata.getPort());
            if (channel == null) {
                LOGGER.info("[Node-{}] Node-{} ({}:{}) is unreachable, skipping.",
                        this.getId(), targetNodeMetadata.getId(), targetNodeMetadata.getAddress(), targetNodeMetadata.getPort());
            } else {
                Api.MQSync(channel, message);
                LOGGER.info("[Node-{}] Record ACK of message {} from server {}.", this.getId(), message.getUuid(), targetNodeMetadata.getId());
                this.getTracker().recordAck(message.getUuid(), message.getLastUpdateServerId(), targetNodeMetadata.getId());
            }
        }
    }

    public boolean propagateMQMessage(final AbstractMessage message) {
//        DebugTool.write("server " + this.getId() + ": propagateMQMessage method");
        // Update local storage
        this.applyMQMessageToStorage(message);

        // Init ACK if it is the message source
//        if (message.getLastUpdateServerId() == this.getId()) {
//            this.getTracker().initProgress(message);
//        }
        // Notify clients
//        this.doMQNotifyClients(message);
//        clients pull data
        final int sourceId = message.getLastUpdateServerId();
        final int currentId = this.getId();

        // this.doSyncWithServer(message, sourceId, currentId);
        this.getExecutorService().submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                try {
                    doMQSyncWithServer(message, sourceId, currentId);
                    return null;
                } catch (Exception exception) {
                    exception.printStackTrace();
                    return null;
                }
            }
        });

        return true;
    }

    public boolean propagateReformMessage(final AbstractMessage message) {
//        DebugTool.write("server " + this.getId() + " propagateReformMessage");

        final int sourceId = message.getLastUpdateServerId();
        final int currentId = this.getId();

        // this.doSyncWithServer(message, sourceId, currentId);
        this.getExecutorService().submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                try {
//                    DebugTool.write("propagateReformMessage submit");
                    doReformSyncWithServer(message, sourceId, currentId);
                    return null;
                } catch (Exception exception) {
                    exception.printStackTrace();
                    return null;
                }
            }
        });

        return true;
    }

    public NodeMetadata findServerById(int serverId) {
        for (NodeMetadata nodeMetadata : this.getNodeList()) {
            if (nodeMetadata.getId() == serverId) {
                return nodeMetadata;
            }
        }
        return null;
    }

    private void doSyncWithServer(AbstractMessage message, int sourceId, int currentId) {
        NodeMetadata source = this.findServerById(sourceId);
        NodeMetadata current = this.findServerById(currentId);
        List<NodeMetadata> initialList = this.getOverlay().calculateNodesToSync(message, source, current);
        Queue<NodeMetadata> sendQueue = new LinkedBlockingDeque<>(initialList);

        // Sync to servers following the overlay
        int sendLoad = 0;
        while (!sendQueue.isEmpty()) {
            // Fault tolerant
            NodeMetadata nodeMetadata = sendQueue.poll();
            if (this.getHealthManager().isAlive(nodeMetadata)) {
                LOGGER.info("[Node-{}] Sync {} with server {}:{}.", this.getId(), message.getType(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                Channel channel = this.getApiServer().findChannel(nodeMetadata.getAddress(), nodeMetadata.getPort());
                if (channel == null) {
                    LOGGER.info("[Node-{}] Node-{} ({}:{}) is unreachable, skipping.",
                            this.getId(), nodeMetadata.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                } else {
                    sendLoad++;
                    Api.sync(channel, message);
                    LOGGER.info("[Node-{}] Record ACK of message {} from server {}.", this.getId(), message.getUuid(), nodeMetadata.getId());
                    this.getTracker().recordAck(message.getUuid(), message.getLastUpdateServerId(), nodeMetadata.getId());
                }
            } else {
                LOGGER.info("[Node-{}] Server {}:{} (id = {}) is unreachable, attempting to send to its children.",
                        this.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort(), nodeMetadata.getId());
                List<NodeMetadata> list = this.getOverlay().calculateNodesToSync(message, source, nodeMetadata);
                sendQueue.addAll(list);
            }
        }

        if (sendLoad > this.getOverlay().getMaxLoad()) {
            this.reform();
        }

    }

    private void doMQSyncWithServer(AbstractMessage message, int sourceId, int currentId) {
//        DebugTool.write("doMQSyncWithServer: node id: " + this.getId()
//                + ", topic: " + message.getTopic() + ", data: " + message.getData().toString());
//        DebugTool.write("[doMQSyncWithServer method] " + "sourceId: " + sourceId + ", currentId: " + currentId);
//        DebugTool.write("size:" + ((StarOverlay)this.getOverlay()).getNodeList().size());
        NodeMetadata source = this.findServerById(sourceId);
        NodeMetadata current = this.findServerById(currentId);
        List<NodeMetadata> initialList = this.getOverlay().calculateNodesToSync(message, source, current);;
        // if (this.getState().equals(APPLIED_STATE)) {
        //     initialList = this.getNewOverlay().calculateNodesToSync(message, source, current);
        // } else {
        //     initialList = this.getOverlay().calculateNodesToSync(message, source, current);
        // }

//        DebugTool.write("[doMQSyncWithServer method] " + " initialList size " + initialList.size());
        Queue<NodeMetadata> sendQueue = new LinkedBlockingDeque<>(initialList);

        // Sync to servers following the overlay
        int sendLoad = 0;

        while (!sendQueue.isEmpty()) {
            sendLoad++;
            // Fault tolerant
            NodeMetadata nodeMetadata = sendQueue.poll();
            if (this.getHealthManager().isAlive(nodeMetadata)) {
                LOGGER.info("[Node-{}] Sync {} with server {}:{}.", this.getId(), message.getType(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                Channel channel = this.getApiServer().findChannel(nodeMetadata.getAddress(), nodeMetadata.getPort());
                if (channel == null) {
                    // DebugTool.write("[doMQSyncWithServer method] " + "channel == null");
                    LOGGER.info("[Node-{}] Node-{} ({}:{}) is unreachable, skipping.",
                            this.getId(), nodeMetadata.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                } else {
                    Api.MQSync(channel, message);
                    LOGGER.info("[Node-{}] Record ACK of message {} from server {}.", this.getId(), message.getUuid(), nodeMetadata.getId());
                    this.getTracker().recordAck(message.getUuid(), message.getLastUpdateServerId(), nodeMetadata.getId());
//                    DebugTool.write("server " + this.getId() + " MQ Sync to server " + nodeMetadata.getId());
                }
            } else {
                LOGGER.info("[Node-{}] Server {}:{} (id = {}) is unreachable, attempting to send to its children.",
                        this.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort(), nodeMetadata.getId());
                List<NodeMetadata> list = this.getOverlay().calculateNodesToSync(message, source, nodeMetadata);
                sendQueue.addAll(list);
            }
        }

        if (sendLoad > this.getOverlay().getMaxLoad()) {
            this.reform();
        }

        if (this.getOverlay() instanceof FicusOverlay && sourceId == currentId) {
            List<List<NodeMetadata>> sendGroupList = ((FicusOverlay) this.getOverlay()).calculateGroupsToSync(message, source, current);

            // Sync to servers following the overlay
            for (List<NodeMetadata> groupNodeList : sendGroupList) {
                for (NodeMetadata nodeMetadata : groupNodeList) {
                    if (this.getHealthManager().isAlive(nodeMetadata)) {
//                LOGGER.info("[Node-{}] Sync {} with server {}:{}.", this.getId(), message.getType(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                        Channel channel = this.getApiServer().findChannel(nodeMetadata.getAddress(), nodeMetadata.getPort());
                        if (channel == null) {
//                    LOGGER.info("[Node-{}] Node-{} ({}:{}) is unreachable, skipping."
//                            , this.getId(), nodeMetadata.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                        } else {
                            Api.MQSync(channel, message);
//                    LOGGER.info("[Node-{}] Record ACK of message {} from server {}.", this.getId(), message.getUuid(), nodeMetadata.getId());
                            this.getTracker().recordAck(message.getUuid(), message.getLastUpdateServerId(), nodeMetadata.getId());
                        }
                        break;
//                    } else {
//                LOGGER.info("[Node-{}] Server {}:{} (id = {}) is unreachable, attempting to send to its children."
//                        , this.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort(), nodeMetadata.getId());
//                        List<NodeMetadata> list = this.getOverlay().calculateNodesToSync(message, source, nodeMetadata);
//                        sendQueue.addAll(list);
                    }
                }
            }
        }
    }

    private void doReformSyncWithServer(AbstractMessage message, int sourceId, int currentId) {
        DebugTool.write("doReformSyncWithServer: node id: " + this.getId());
        NodeMetadata source = this.findServerById(sourceId);
        NodeMetadata current = this.findServerById(currentId);
        Overlay overlay;
        if (message.getType().equals(Constants.REFORM_PREPARE)) {
            overlay = this.getOverlay();
        } else {
            overlay = this.getNewOverlay();
        }
        List<NodeMetadata> initialList = overlay.calculateNodesToSync(message, source, current);
        Queue<NodeMetadata> sendQueue = new LinkedBlockingDeque<>(initialList);

        // Sync to servers following the overlay
        while (!sendQueue.isEmpty()) {
            // Fault tolerant
            NodeMetadata nodeMetadata = sendQueue.poll();
            if (this.getHealthManager().isAlive(nodeMetadata)) {
//                LOGGER.info("[Node-{}] Sync {} with server {}:{}.", this.getId(), message.getType(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                Channel channel = this.getApiServer().findChannel(nodeMetadata.getAddress(), nodeMetadata.getPort());
                if (channel == null) {
//                    LOGGER.info("[Node-{}] Node-{} ({}:{}) is unreachable, skipping."
//                            , this.getId(), nodeMetadata.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort());
                } else {
                    Api.reformSync(channel, message);
//                    LOGGER.info("[Node-{}] Record ACK of message {} from server {}.", this.getId(), message.getUuid(), nodeMetadata.getId());
                    this.getTracker().recordAck(message.getUuid(), message.getLastUpdateServerId(), nodeMetadata.getId());
                    DebugTool.write("server " + this.getId() + " send " + message.getType() + " to server " + nodeMetadata.getId());
                }
            } else {
//                LOGGER.info("[Node-{}] Server {}:{} (id = {}) is unreachable, attempting to send to its children."
//                        , this.getId(), nodeMetadata.getAddress(), nodeMetadata.getPort(), nodeMetadata.getId());
                List<NodeMetadata> list = overlay.calculateNodesToSync(message, source, nodeMetadata);
                sendQueue.addAll(list);
            }
        }

    }

    private void doNotifyClients(AbstractMessage message) {
        Item item = new Item(message.getApplicationName(), message.getKey());
        if (this.getSubscription().containsKey(item)) {
            Set<ClientMetadata> clients = this.getSubscription().get(item);
            this.getClientDispatcher().notifyClients(clients, message);
        }
    }

    private void doMQNotifyClients(AbstractMessage message) {
        String topic = message.getTopic();
        if (this.getMQSubscription().containsKey(topic)) {
            Set<ClientMetadata> clients = this.getMQSubscription().get(topic);

            this.getClientDispatcher().notifyClients(clients, message);
        }
    }

    private void applyMessageToStorage(AbstractMessage message) {
        String applicationName = message.getApplicationName();
        String key = message.getKey();
        Item item = this.getStorage().getItemService().getItem(applicationName, key);
        if (item == null) {
            this.getStorage().getItemService().newItem(applicationName, key);
        }
        try {
            FileWriter fileWriter = new FileWriter("track.txt");
            fileWriter.write("server " + this.getId() + " store message from " + message.getFromWhere());
            fileWriter.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        this.getStorage().getMessageService().newMessage(message);
    }

    private void applyMQMessageToStorage(AbstractMessage message) {
        String topic = message.getTopic();
//        this.getStorage().getMessageService().newMessage(message);
        // if (!messageQueue.containsKey(topic)) {
        //     messageQueue.put(topic, new ArrayList<>());
        // }
        // messageQueue.get(topic).add(message.getData());
//        DebugTool.write(String.valueOf(this.getId()), "{[topic:" + topic + "], [id:" +  this.getId() + "], [data_string" + message.getData().toString() + "]}");

//        if (this.getMQStorage().getTopicService().getTopic(topic) == null) {
//            this.getMQStorage().getTopicService().newTopic(topic);
//        }
//        this.getMQStorage().getMessageService().newMessage(message);
        if (this.getStorage().getTopicService().getTopic(topic) == null) {
            this.getStorage().getTopicService().newTopic(topic);
        }
        this.getStorage().getMessageService().newMessage(message);

        // switch (topic) {
        //     case REFORM_TOPIC:
        //         if (this.getState().equals(NORMAL_STATE) ||
        //             (this.getState().equals(PROPOSER_STATE) && message.getLastUpdate().compareTo(this.getProposeDate()) == -1 )) {
        //             this.setState(FOLLOWER_STATE);
        //             this.setNewNodeList(NodeMetadata.ListFromString(message.getData().toString()));
        //             this.setNewOverlay(new TreeOverlay(((TreeOverlay)this.getOverlay()).getBranch()));
        //             this.getNewOverlay().buildOverlay(this.getNewNodeList());
        //             this.responseCount = 0;
        //         } else {
        //             this.push(PROPOSE_RESPONSE, new Data(String.valueOf(message.getLastUpdateServerId())));
        //         }
        //         break;
        //     case PROPOSE_RESPONSE:
        //         if (this.getState().equals(PROPOSER_STATE) 
        //             && Integer.parseInt(message.getData().toString()) == this.getId()) {
        //             if (this.responseCount < this.getNewNodeList().size()) {
        //                 this.responseCount++;
        //             } else {
        //                 this.setState(LEADER_STATE);
        //                 this.push(APPLY_REQUEST, new Data(this.getAddress()));
        //             }
        //         }
        //         break;
        //     case APPLY_REQUEST:
        //         if (this.getState().equals(FOLLOWER_STATE) 
        //             && Integer.parseInt(message.getData().toString()) == this.getId()) {
        //             this.setState(APPLIED_STATE);
        //             this.push(APPLY_RESPONSE, new Data(this.getAddress()));
        //         }
        //         break;
        //     case COMMIT_REQUEST:
        //         this.setOverlay(this.getNewOverlay());
        //         this.setState(NORMAL_STATE);
        //         break;
        //     default:
        //         break;
        // }
    }

    public Map<String, ArrayList<Data>> getMessageQueue() {
        return messageQueue;
    }

    public void setMessageQueue(Map<String, ArrayList<Data>> messageQueue) {
        this.messageQueue = messageQueue;
    }

    private Map<String, ArrayList<Data>> messageQueue;

    public synchronized boolean start() {
        if (this.isRunning()) {
            return true;
        }
        try {
            this.setApiServer(new NettyServer(this, this.getApiPort()));
            this.getApiServer().start();
            this.setApiPort(this.getApiServer().getPort());

            this.setUiServer(new Server());
            ServerConnector serverConnector = new ServerConnector(this.getUiServer());
            serverConnector.setPort(this.getUiPort());
            this.getUiServer().setConnectors(new Connector[]{serverConnector});
            ServletContextHandler servletContextHandler = new ServletContextHandler();
            this.registerHandlers(servletContextHandler);
            this.getUiServer().setHandler(servletContextHandler);
            this.getUiServer().start();
            this.setAddress(InetAddress.getLocalHost().getHostAddress());
            this.setUiPort(serverConnector.getLocalPort());
            if (this.getExecutorService() == null || this.getExecutorService().isShutdown()) {
                this.setExecutorService(Executors.newCachedThreadPool());
            }

            this.setRunning(true);
            return true;
        } catch (Exception exception) {
            exception.printStackTrace();
            return false;
        }
    }

    public synchronized boolean stop() {
        if (!this.isRunning()) {
            return true;
        }

        try {
            this.getApiServer().stop();
            this.getUiServer().stop();
//            this.getWorkingThread().interrupt();
            Thread workingThread = this.getWorkingThread();
            if (workingThread != null) {
                workingThread.interrupt();
            }

            this.executorService.shutdown();

            this.setRunning(false);
            return true;
        } catch (Exception exception) {
            exception.printStackTrace();
            return false;
        }
    }

    public synchronized void subscribe(String callbackAddress, int callbackPort, String applicationName, String key) {
        LOGGER.info("[Node-{}] subscribe() called: Callback Address = {}, Callback Port = {}, Application Name = {}, Key = {}.",
                this.getId(), callbackAddress, callbackPort, applicationName, key);
        Item item = new Item(applicationName, key);
        if (this.getSubscription().get(item) == null) {
            this.getSubscription().put(item, new HashSet<>());
        }
        Set<ClientMetadata> subscribers = this.getSubscription().get(item);
        ClientMetadata clientMetadata = new ClientMetadata(callbackAddress, callbackPort);
        if (!subscribers.contains(clientMetadata)) {
            subscribers.add(clientMetadata);
        }
        this.getApiServer().connect(clientMetadata.getAddress(), clientMetadata.getPort());
        this.getConnectedClients().add(clientMetadata);
    }

    public synchronized void unsubscribe(String callbackAddress, int callbackPort, String applicationName, String key) {
        LOGGER.info("[Node-{}] unsubscribe() called: Callback Address = {}, Callback Port = {}, Application Name = {}, Key = {}.",
                this.getId(), callbackAddress, callbackPort, applicationName, key);
        Item item = new Item(applicationName, key);
        if (this.getSubscription().get(item) == null) {
            return;
        }
        Set<ClientMetadata> subscribers = this.getSubscription().get(item);
        ClientMetadata clientMetadata = new ClientMetadata(callbackAddress, callbackPort);
        if (subscribers.contains(clientMetadata)) {
            subscribers.remove(clientMetadata);
        }

        this.cleanUpConnectedClients(clientMetadata);
    }

    public synchronized void subscribe(String callbackAddress, int callbackPort, String topic) {
        LOGGER.info("[Node-{}] subscribe() called: Callback Address = {}, Callback Port = {}, Topic = {}.",
                this.getId(), callbackAddress, callbackPort, topic);
        this.getMQSubscription().computeIfAbsent(topic, k -> new HashSet<>());
        Set<ClientMetadata> subscribers = this.getMQSubscription().get(topic);
        ClientMetadata clientMetadata = new ClientMetadata(callbackAddress, callbackPort);
//        if (!subscribers.contains(clientMetadata)) {
//            subscribers.add(clientMetadata);
//        }
        subscribers.add(clientMetadata);
        this.getApiServer().connect(clientMetadata.getAddress(), clientMetadata.getPort());
        this.getConnectedClients().add(clientMetadata);
    }

    public synchronized void unsubscribe(String callbackAddress, int callbackPort, String topic) {
        LOGGER.info("[Node-{}] unsubscribe() called: Callback Address = {}, Callback Port = {}, Topic = {}.",
                this.getId(), callbackAddress, callbackPort, topic);
        if (this.getMQSubscription().get(topic) == null) {
            return;
        }
        Set<ClientMetadata> subscribers = this.getMQSubscription().get(topic);
        ClientMetadata clientMetadata = new ClientMetadata(callbackAddress, callbackPort);
//        if (subscribers.contains(clientMetadata)) {
//            subscribers.remove(clientMetadata);
//        }
        subscribers.remove(clientMetadata);
        this.cleanUpConnectedClients(clientMetadata);
    }

    private void cleanUpConnectedClients(ClientMetadata clientMetadata) {
        boolean exist = false;
        for (Set<ClientMetadata> clients : this.getSubscription().values()) {
            if (clients.contains(clientMetadata)) {
                exist = true;
                break;
            }
        }
        if (!exist) {
            this.getConnectedClients().remove(clientMetadata);
        }
    }

    public void initCluster(List<NodeMetadata> nodeList) {
        this.setNodeList(nodeList);
        this.getHealthManager().init();
        this.getOverlay().buildOverlay(this.getNodeList());
        this.setWorkingThread(new Thread(this.getWorker()));
        this.getWorkingThread().start();
        this.reconnect(nodeList);
    }

    public synchronized void reconnect(List<NodeMetadata> nodeList) {
        if (DebugTool.flag == 0 && nodeList.size() == 0) {
            DebugTool.write("[Node] node " + this.getId()
                    + " reconnect " + NodeMetadata.ListToString(this.getNodeList()));
            DebugTool.flag = 1;
        }
        for (NodeMetadata metadata : nodeList) {
            // if (metadata.getId() > this.getId()) {
            Channel channel = this.getApiServer().findChannel(metadata.getAddress(), metadata.getPort());

            if (channel == null) {
//                DebugTool.write("[Node] reconnect: channel == null");
                try {
                    LOGGER.info("[Node-{}] Reconnecting to Node-{} ({}:{}).",
                            this.getId(), metadata.getId(), metadata.getAddress(), metadata.getPort());

                    this.getApiServer().connect(metadata.getAddress(), metadata.getPort());
                } catch (Exception exception) {
                    exception.printStackTrace();
                }
            }
            // }
        }
    }

    public void updateCpuLoad(long delay) {
        SystemInfo si = new SystemInfo();
        HardwareAbstractionLayer hal = si.getHardware();
        CentralProcessor cpu = hal.getProcessor();
        double load = cpu.getSystemCpuLoad(delay);
        this.setCurrentCpuLoad(load);
    }

    public void cleanDB() {
        this.getStorage().cleanDB();
    }

    public void reformMessageHandler(ReformMessage reformMessage) {
        if (!(this.getOverlay() instanceof ReformOverlay)) {
            return;
        }
        String action = ((ReformOverlay) this.getOverlay()).reformMessageHandler(reformMessage, this.getId());

        switch (action) {
            case ReformOverlay.CONTINUE:
                this.propagateMessage(reformMessage);
                break;
            case ReformOverlay.RECORD:
                // this.getOverlay().setNewNodeList(null);
                this.propagateMessage(reformMessage);
                break;
            case ReformOverlay.APPLY_COMMIT:
                // this.getOverlay().setNewNodeList(null);
                this.propagateMessage(reformMessage);
                break;
            case ReformOverlay.APPLY_DEPRECATE:
                // this.getOverlay().setNewNodeList(null);
                this.propagateMessage(reformMessage);
                break;
            default:
                break;
        }

    }

    public boolean equals(NodeMetadata nodeMetadata) {
        return (this.getId() == nodeMetadata.getId())
                && this.getAddress().equals(nodeMetadata.getAddress())
                && (this.getApiPort() == nodeMetadata.getPort());
    }
}
